package cn.spring.rabbitmq.demo.rabbitmq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.spring.rabbitmq.demo.rabbitmq.message.DemoMessage;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Resource;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @Author tanyong
 * @Version MessageConsumer v1.0.0 2025/9/12 15:10 $$
 */
@Slf4j
@Component
public class MessageConsumer {

    @Resource
    private ObjectProvider<MessageConverter> messageConverterObjectProvider;

    /**
     * 监听并处理DemoMessage类型的消息
     *
     * @param message 消息内容
     */
    @RabbitListener(queuesToDeclare = {@Queue("demo.queue")})
    public void handleMessageByAnnotation(DemoMessage message) {
        log.info("[handleMessageByAnnotation] 收到消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 监听广播消息1
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout.queue.1"),
            exchange = @Exchange(value = "fanout-exchange", type = "fanout")
    ))
    public void handleFanoutMessage1(DemoMessage message) {
        log.info("[handleFanoutMessage1] 收到广播消息: userId={}, message={},  createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }


    /**
     * 监听广播消息2
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout.queue.2"),
            exchange = @Exchange(value = "fanout-exchange", type = "fanout")
    ))
    public void handleFanoutMessage2(DemoMessage message) {
        log.info("[handleFanoutMessage2] 收到广播消息: userId={}, message={},  createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 监听topic消息1
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue.1"),
            exchange = @Exchange(value = "topic-exchange", type = "topic"),
            key = "topic.message.specific"
    ))
    public void handleTopicMessage1(DemoMessage message) {
        log.info("[handleTopicMessage1] 收到Topic消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 监听topic消息2
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue.2"),
            exchange = @Exchange(value = "topic-exchange", type = "topic"),
            key = "topic.message.*"
    ))
    public void handleTopicMessage2(DemoMessage message) {
        log.info("[handleTopicMessage2] 收到Topic消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 监听延迟消息处理队列
     * 当延迟队列中的消息过期后，会被自动转发到此队列进行处理
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("delay.process.queue"),
            exchange = @Exchange(value = "delay.process.exchange", type = "direct"),
            key = "delay.process"
    ))
    public void handleDelayMessage(DemoMessage message) {
        log.info("[handleDelayMessage] 收到延迟消息: userId={}, message={}, createTime={}",
                message.getUserId(), message.getMessage(), message.getCreateTime());
    }

    /**
     * 批量处理消息
     *
     * @param messages
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("batch.queue"),
            exchange = @Exchange(value = "batch.exchange", type = "direct"),
            key = "batch"
    ), batch = "true", containerFactory = "batchListenerContainerFactory", ackMode = "MANUAL")
    public void handleBatchMessage(List<Message> messages, Channel channel) {
        log.info("[handleBatchMessage] 开始处理批量消息，共 {} 条", messages.size());

        if (CollUtil.isEmpty(messages)) {
            log.info("[handleBatchMessage] 消息列表为空，无需处理");
            return;
        }

        // 分别存储成功和失败的消息
        List<Message> successMessages = new ArrayList<>();
        List<Message> failedMessages = new ArrayList<>();

        // 批量转换消息
        for (Message message : messages) {
            try {
                DemoMessage demoMessage = (DemoMessage) messageConverterObjectProvider.getObject().fromMessage(message);
                demoMessage.setDeliveryTag(message.getMessageProperties().getDeliveryTag());
                successMessages.add(message);
                log.debug("[handleBatchMessage] 消息转换成功: deliveryTag={}", message.getMessageProperties().getDeliveryTag());
            } catch (Exception e) {
                log.error("[handleBatchMessage] 消息转换失败: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                failedMessages.add(message);
            }
        }

        // 处理成功转换的消息
        if (CollUtil.isNotEmpty(successMessages)) {
            try {
                log.info("[handleBatchMessage] 开始处理 {} 条成功转换的消息", successMessages.size());
                // 模拟处理时间 - 实际应用中这里应该是真正的业务逻辑
                processMessages(successMessages);

                // 批量确认所有成功处理的消息
                for (Message message : successMessages) {
                    try {
                        long deliveryTag = message.getMessageProperties().getDeliveryTag();
                        channel.basicAck(deliveryTag, false);
                        log.debug("[handleBatchMessage] 消息确认成功: deliveryTag={}", deliveryTag);
                    } catch (IOException e) {
                        log.error("[handleBatchMessage] 消息确认失败: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                    }
                }
                log.info("[handleBatchMessage] 成功处理并确认 {} 条消息", successMessages.size());

            } catch (Exception e) {
                log.error("[handleBatchMessage] 处理消息时发生异常", e);
                // 如果处理过程中出现异常，将所有成功转换的消息标记为失败
                failedMessages.addAll(successMessages);
            }
        }

        // 处理转换失败的消息
        if (!failedMessages.isEmpty()) {
            log.warn("[handleBatchMessage] 共 {} 条消息处理失败，尝试重新入队", failedMessages.size());
            for (Message message : failedMessages) {
                try {
                    long deliveryTag = message.getMessageProperties().getDeliveryTag();
                    // 第三个参数设为true，表示将消息重新放回队列
                    channel.basicNack(deliveryTag, false, true);
                    log.debug("[handleBatchMessage] 消息重新入队: deliveryTag={}", deliveryTag);
                } catch (IOException e) {
                    log.error("[handleBatchMessage] 消息重新入队失败: deliveryTag={}", message.getMessageProperties().getDeliveryTag(), e);
                }
            }
        }

        log.info("[handleBatchMessage] 批量消息处理完成: 成功={}条, 失败={}条",
                successMessages.size(), failedMessages.size());
    }

    /**
     * 实际处理消息的方法
     * 在实际应用中，这里应该包含真正的业务逻辑
     *
     * @param messages 待处理的消息列表
     * @throws Exception 处理异常
     */
    private void processMessages(List<Message> messages) throws Exception {
        // 模拟处理时间
        Thread.sleep(50L);

        // 实际应用中，这里应该是真正的业务逻辑处理
        log.info("[processMessages] 处理了 {} 条消息", messages.size());
    }
}
